package cn.hloger.spider.demo.webSocketDemo.handle;

import cn.hloger.spider.demo.webSocketDemo.bean.RwZxRzVO;
import com.alibaba.fastjson.JSON;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import javax.websocket.Session;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.PriorityBlockingQueue;

/**
 * 构建此类的原因是任务执行的速度过快，前端websocket还未打开，导致消息无法推送到前端
 * 因此构建消息缓存队列，暂时缓存消费的信息，并持久化1分钟，如果1分钟之后websocket没有连接，则释放缓存队列
 */
@Data
@Slf4j
public class WsMessageHandle {

    //定义最长的信息条数为10万条，避免内存溢出，后续需求有需要可以调整队列消息长度
    private static final Integer MAX_LENGTH = 100000;

    /**
     * 最后插入数据时间
     */
    private LocalDateTime lastInsertTime;

    /**
     * 任务唯一标识id，一般来说是sqlid
     */
    private String taskId;

    /**
     * 消息缓存队列
     */
    private PriorityBlockingQueue<RwZxRzVO> queue;

    /**
     * 发送消息线程
     */
    private Future future;

    /**
     * 是否处于连接状态
     * <p>
     * 连接状态的handle会一直从队列中获取信息，并且阻塞执行
     * 处于连接状态的handle不能被自动清除，只能调用destroy方法
     */
    private boolean connect;


    public WsMessageHandle(String taskId) {
        this.taskId = taskId;
        //采用升序排列
        this.queue = new PriorityBlockingQueue<>(MAX_LENGTH, (RwZxRzVO o1, RwZxRzVO o2) -> o1.getYxsj().compareTo(o2.getYxsj()));
        lastInsertTime = LocalDateTime.now();
        connect = false;
    }

    /**
     * 保存信息
     *
     * @param msg
     */
    public void save(RwZxRzVO msg) {
        //如果插入成功，更新最后插入时间，如果插入失败，则队列已满，直接丢弃msg
        if (queue.offer(msg)) {
            lastInsertTime = LocalDateTime.now();
        } else {
            log.error("队列已满丢失taskid:{}的消息：{}", taskId, msg);
        }
    }

    /**
     * 执行的websocket发送消息，如果队里中没有内容则阻塞等待
     * 待前端websocket关闭，则调用destory方法停止线程
     *
     * @param session         websocket对象
     * @param executorService 线程池
     */
    public void sendMsg(Session session, ExecutorService executorService) {
        synchronized (this) {
            //如果当前任务已经开启发送消息线程，则跳过
            //没有则从线程池中分配线程开始发送
            if (connect) {
                return;
            }

            this.future = executorService.submit(() -> {
                try {
                    int count = 0;//为防止出现死循环，定义一个变量，循环最大次数为100000
                    while (count < MAX_LENGTH) {
                        RwZxRzVO msg = queue.take();
                        session.getBasicRemote().sendText(JSON.toJSONString(msg));
                        count++;
                    }
                } catch (InterruptedException e) {
                    log.error(e.getMessage());
                    //如果出现中断异常，则直接中断当前子线程
                    Thread.currentThread().interrupt();
                } catch (IOException io) {
                    log.error(io.getMessage());
                }
            });
            this.connect = true;
        }
    }

    /**
     * 释放缓存队列
     */
    public void destory() {
        if (this.future != null) {
            this.future.cancel(true);//中断阻塞线程
        }
        queue.clear();//清除队列内容
        connect = false;
    }
}
